package com.deepglint.sink;

import com.deepglint.beans.SensorReading;
import com.deepglint.source.SourceTest_UDF;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-16 23:23
 */
public class SinkTest_JDBC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String inputPath = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStream<String> inputStream = env.readTextFile(inputPath);

        // 转换成SensorReading类型
//        DataStream<SensorReading> streamMap = inputStream.map(value -> {
//            String[] split = value.split(",");
//            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
//        });
        DataStreamSource<SensorReading> streamSource = env.addSource(new SourceTest_UDF.MySensorSource());

        // 连接pg数据库
        streamSource.addSink(new MyJDBCSink());

        env.execute();
    }

    // 实现自定义的sink
    static class MyJDBCSink extends RichSinkFunction<SensorReading> {
        // 声明连接的预编译
        private Connection connection = null;
        PreparedStatement insertStatement = null;
        PreparedStatement UpdateStatement = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/fink_test", "postgres", "111");
            insertStatement = connection.prepareStatement("insert into sensor_temp(id,temp) values (?,?)");
            UpdateStatement = connection.prepareStatement("update sensor_temp set temp = ? where id = ?");
        }

        // 每来一条数据，调用连接，执行sql
        @Override
        public void invoke(SensorReading value, Context context) throws Exception {
            // 直接更新语句，如果没有更新就插入
            UpdateStatement.setDouble(1,value.getTemperature());
            UpdateStatement.setString(2,value.getId());
            UpdateStatement.execute();
            if (UpdateStatement.getUpdateCount() == 0){
                insertStatement.setString(1,value.getId());
                insertStatement.setDouble(2,value.getTemperature());
                insertStatement.execute();
            }
        }

        @Override
        public void close() throws Exception {
            connection.close();
            UpdateStatement.close();
            insertStatement.close();
        }
    }
}
